package com.telra.app.server2.cassandra;

import com.telra.app.server.data.OrderEvent;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *  Writes OrderEvent objects to Cassandra asynchronously
 */
public class CassandraWriter implements CassandraEnabled {
    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(CassandraWriter.class);

    private Keyspace keyspace;
    private BlockingQueue<OrderEvent> queue;
    private Insertor ins;
    private static StringSerializer stringSerializer = StringSerializer.get();

    public CassandraWriter() {

        queue = new LinkedBlockingQueue<OrderEvent>();

        ExecutorService exec = Executors.newSingleThreadExecutor();
        ins = new Insertor(1000, 10000);
        exec.execute(ins);
    }
    
    public void cancel() {
       ins.cancel();
    }

    public Keyspace getKeyspace() {
        return keyspace;
    }

    public void setKeyspace(Keyspace keyspace) {
        this.keyspace = keyspace;
    }

    public void insertOrder(OrderEvent order) {
        try {
            queue.put(order);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private class Insertor implements Runnable {

        private volatile boolean cancelled;
        private int sleep;
        private int queueSize;

        public Insertor(int sleep, int queueSize) {
            this.sleep = sleep;
            this.queueSize = queueSize;
        }

        public void run() {

            while (!cancelled) {
                try {
                    Thread.sleep(sleep);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                List<OrderEvent> orders = new ArrayList<OrderEvent>();
                queue.drainTo(orders);

                log.info("queue:" + orders.size());

                Mutator<String> mutator = HFactory.createMutator(keyspace, stringSerializer);

                int c= 0;
                for (OrderEvent order : orders) {
                    //insert(key, colFamily, column<key,val>)
                    mutator.
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("itemName", order.getItemName())).
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("price", String.valueOf(order.getPrice()))).
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("category", order.getCategory().toString()));

                    if (c++ > queueSize) {
                        mutator.execute();
                        c = 0;
                    }
                }

                if (c > 0) {	//flush the rest!
                    mutator.execute();
                }

                log.info("queue processed");
            }
        }

        public void cancel() {
            cancelled = true;
        }

        public boolean isCancelled() {
            return cancelled;
        }

    }

}